package cn.pugle.base.net.socket;

import cn.pugle.base.net.socket.client.CONS;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 使用NIO的Server 非阻塞， Reactor模式 = Selector <br/> Reactor 仍是同步IO。 Proactor异步 事件回调 <br/>
 * Selector(感觉有两个地方涉及用不用selector来多路复用，一个是ServerSocket accept的时候，一个是接收到的SocketChannel读数据的时候 ) <br/>
 *
 * 仿照Doug Lea nio中线程池和两个Selector版 <br/>
 * <pre>
 * //xxx 真尼玛蠢啊，两个Reactor线程互操作被锁住了，为什么不想想其他的线程间通信方法呢。 <br/>
 * // 参考zk的NIOCnxnFactory类的代码:
 * // 1. AcceptThread，调用SelectorThread的方法 //xxx
 *   （将新连接的socket放到SelectorThread的队列中，然后调用wakeup，SelectorThread在select被wakeup后，自己去队列里拿新连接，注册）
 * // 2. 多个SelectorThread，轮询的方式添加连接。
 * // 3. SelectorThread返回可读时，cnxn.disableSelectable();
 *    key.interestOps(0);来禁止这个key继续返回可读。doWork处理之后再恢复。（前一次）
 * // 4. 使用线程池来doIO而不是在SelectorThread中进行，但是线程池不应该知道自己要执行的是什么东西，所以有IORequest类
 *    实现WorkService的hook。
 * // 5. 不用也不应该!去保证同一个连接的多次读是由一个线程进行的!, 那么如何保证多次的数据能在一起处理呢，ZK中是在SelectionKey上attach了一个NIOServerCnxn类维护数据、doIO
 * </pre>
 *
 * @author tzp
 * @date 15-12-11
 */
public class TwoReactorServer {

    private ServerSocketChannel ssc = null;
    private MainReactorWorker mainReactor = null;
    private SubReactorWorker subReactor = null;
    private InetSocketAddress inetAddress;

    private ExecutorService ioThreadPool;


    public TwoReactorServer() {
        inetAddress = new InetSocketAddress(CONS.SERVER_ADDRESS, CONS.SERVER_PORT);
        ioThreadPool = Executors.newFixedThreadPool(10);
    }

    public static void main(String[] args) {
        new TwoReactorServer().bootstrap();
    }

    public void bootstrap() {
        try {
            ssc = ServerSocketChannel.open();
            ssc.bind(inetAddress);
            ssc.configureBlocking(false);
            System.out.println("Open ServerSocketChannel and bind " + inetAddress);

            mainReactor = new MainReactorWorker();
            Thread mainReactorThread = new Thread(mainReactor);//主Selector
            mainReactorThread.start();
            System.out.println("MainReactor started");

            subReactor = new SubReactorWorker();
            Thread subReactorThread = new Thread(subReactor);//副Selector
            subReactorThread.start();
            System.out.println("SubReactor started");

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 主Reactor线程, 负责接受新的连接请求. 将新的连接请求发给subReactor
     */
    class MainReactorWorker implements Runnable {
        private Selector mainSelector;

        /**
         * 在构造的时候打开selector, 注册ssc
         */
        public MainReactorWorker() {
            try {
                mainSelector = Selector.open();
                ssc.register(mainSelector, SelectionKey.OP_ACCEPT);//向主Selector注册 ServerChannel，关心连接accept状态
                System.out.println("MainReactor register OP_ACCEPT " + ssc.getLocalAddress().toString());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            try {
                Thread.sleep(15 * 1);//xxx 睡15s等我启两个client.下面的selectedKeys返回一个key, 这个key的ssc可以accept()两个sc
                while (!Thread.interrupted()) {
                    mainSelector.select();
                    Set<SelectionKey> selectionKeys = mainSelector.selectedKeys();
                    for (SelectionKey sk : selectionKeys) {
                        System.out.println("MainReactor select a key, its interestOps: " + sk.interestOps());//sk 是int位
                        ServerSocketChannel serverChannel = (ServerSocketChannel) sk.channel();//SelectionKey.accept()返回的是原来的ssk的serverSocket
                        SocketChannel channel = serverChannel.accept();
                        doAccept(channel);
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        /**
         * 处理主Reactor接受到的线程
         */
        private void doAccept(SocketChannel channel) {
            try {
                channel.configureBlocking(false);

//                System.out.println("haha");// 使用多线程 原来卡再下面这句，由于SubReactorWorker的线程执行了subReactor的select方法获得了锁？
//                channel.register(subSelector, SelectionKey.OP_READ);
//                System.out.println("hehe");

                subReactor.addAcceptedConnection(channel);
                System.out.println("Add a channel to SubReactor's queue " + channel.getRemoteAddress());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 子Reactor线程,负责处理连接发送数据的请求.
     */
    class SubReactorWorker implements Runnable {
        private final Queue<SocketChannel> acceptedQueue;//这个队列是给accept线程往里放接受到的SocketChannel用的.
        private Selector subSelector;

        public SubReactorWorker() {
            acceptedQueue = new LinkedBlockingQueue<>();
            try {
                subSelector = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                doSelect();
                processAcceptedConnections();
            }
        }

        private void doSelect() {
            try {
                subSelector.select();
                System.out.println("+ SubReactor select return");
                Set<SelectionKey> selectionKeys = subSelector.selectedKeys();
                for (SelectionKey sk : selectionKeys) {
                    System.out.println("+ SubReactorWorker select a key, its interestOps: " + sk.interestOps());
                    if (sk.isReadable() || sk.isWritable()) {
                        doIO(sk);//select返回了, 线程正在doIO, 这个时候客户端发来数据会在channel里, 还能读到. 更好理解selector的作用
                    }
                }
                selectionKeys.clear();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }


        /**
         * Place new accepted connection onto a queue for adding. wakeup selector to do next {@link
         * #processAcceptedConnections}.
         * Do this so only the selector thread modifies what keys are registered with the selector.
         */
        public boolean addAcceptedConnection(SocketChannel accepted) {
            if (!acceptedQueue.offer(accepted)) {
                return false;
            }
            subSelector.wakeup();
            return true;
        }

        /**
         * Iterate over the queue of accepted connections that have been
         * assigned to this thread but not yet placed on the selector.
         */
        private void processAcceptedConnections() {
            SocketChannel acceptedChannel;
            while ((acceptedChannel = acceptedQueue.poll()) != null) {
                SelectionKey selectionKey;
                try {
                    selectionKey = acceptedChannel.register(subSelector, SelectionKey.OP_READ);//没有deregister. 用key.cancel
                    MySocketChannelWorkspace workspace = new MySocketChannelWorkspace(acceptedChannel, selectionKey);
                    selectionKey.attach(workspace);
                    subSelector.wakeup();
                    System.out.println("+ SubReactor register a channel OP_READ and attach workspace" + acceptedChannel.getRemoteAddress());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private void doIO(SelectionKey sk) {
            // Stop selecting this key while processing on its
            // connection
            sk.interestOps(0);

            MySocketChannelWorkspace channelWorkspace = (MySocketChannelWorkspace) sk.attachment();
            System.out.println("+ begin to doIO for " + channelWorkspace);
            ioThreadPool.execute(new MyChannelReader(channelWorkspace));
            System.out.println("+ thread pool working. subReactorWorker return");
        }
    }

    /**
     * 一个连接(SocketChannel)的工作区, 包含已收到的数据和需要的操作等.
     */
    class MySocketChannelWorkspace {
        private SelectionKey sk;//监听的selectionKey
        private SocketChannel channelToTreat;//连接的channel
        private StringBuilder stringBuilder = new StringBuilder();//存数据用的
        private ByteBuffer buffer = ByteBuffer.allocate(1000);//这里假定 使用的方式是 将客户端发送的数据全读入buffer

        public MySocketChannelWorkspace(SocketChannel channelToTreat, SelectionKey sk) {
            this.channelToTreat = channelToTreat;
            this.sk = sk;
        }

        public void receiveMessage(String message) {
            System.out.println("++ receive " + message);
            stringBuilder.append(message);
        }


        public void readChannel() {
            try {
                int readNum = -2;//啥也不是, 只是不想用do/while.
                while (stringBuilder.lastIndexOf(CONS.PROTOCOL_TAIL) == -1 && readNum != -1 && readNum != 0) {
                    // 前两个条件说明channel关闭了.最后一个说明数据读完了?

                    readNum = channelToTreat.read(buffer);
                    System.out.println("++ " + readNum + " bytes read");
                    int curPos = buffer.position();
                    System.out.println("-- " + curPos + " current position");
                    buffer.flip();
                    byte[] dst = new byte[curPos];
                    buffer.get(dst);

                    String strRece = new String(dst, CONS.CHARSET_NAME);
                    receiveMessage(strRece);

                    buffer.clear();
                }
                if (stringBuilder.lastIndexOf(CONS.PROTOCOL_TAIL) != -1 || readNum == -1) {
                    System.out.println("++ channel end. cancel sk." + this);
                    channelToTreat.close();
                    sk.cancel();
                } else if (readNum == 0) {
                    sk.interestOps(SelectionKey.OP_READ);
                    sk.selector().wakeup();
                    System.out.println(sk.isValid());
                    System.out.println("++ read all data. not end. re select." + this);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public String toString() {
            return "MySocketChannelWorkspace{" +
//                    channelToTreat.getRemoteAddress().toString() +
                    "channelToTreat=" + channelToTreat;
        }
    }

    /**
     * a simple wrapper to do MySocketChannelWorkspace's read opt in thread pool.
     */
    class MyChannelReader implements Runnable {
        private MySocketChannelWorkspace workspace;

        public MyChannelReader(MySocketChannelWorkspace workspace) {
            this.workspace = workspace;
        }

        @Override
        public void run() {
            workspace.readChannel();//因为reactor+nio也是同步io, 所以需要读操作完成后调用处理数据的操作.
//            workspace dothing
        }
    }
}
